package net.bwie.realtime.dy.dwd.log.funcation.zl.realtimegiftfuncation;

import net.bwie.realtime.dy.dwd.log.bean.RoomGiftMoney;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：GiftProcessFuncation
 * @Author: ZddddQ
 * @Date: 2025/5/28 14:41
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class GiftProcessFuncation
    extends ProcessWindowFunction<Double, RoomGiftMoney, String, TimeWindow> {
    private final FastDateFormat dateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void process(String s, Context context, Iterable<Double> elements, Collector<RoomGiftMoney> out) throws Exception {
        // 获取聚合结果
        Double sum = elements.iterator().next();

        // 获取窗口时间并格式化
        TimeWindow window = context.window();
        String windowStart = dateFormat.format(window.getStart());
        String windowEnd = dateFormat.format(window.getEnd());

        //  构建结果对象
        RoomGiftMoney roomGiftMoney = new RoomGiftMoney();
        roomGiftMoney.setRoomId(s);
        roomGiftMoney.setGiftMoney(sum);
        roomGiftMoney.setWindowStartTime(windowStart);
        roomGiftMoney.setWindowEndTime(windowEnd);
        roomGiftMoney.setTs(System.currentTimeMillis());
        out.collect(roomGiftMoney);


    }
}
